#include "model/SubscriberNode.h"

namespace zeroplus::mq
{
    bool SubscriberNode::Run(const OnMsgRecvCb &processMsgCb)
    {
        if (processMsgCb_)
        {
            LOG_ERROR("Has register");
            return false;
        }
        auto handler = [this]()
        {
            while (true)
            {
                auto message = Recv();
                if (message.to_string() == topic_)
                {
                    LOG_DEBUG("前缀数据是topic, 舍弃");
                    // 前缀数据是topic, 舍弃
                    continue;
                }
                processMsgCb_(message);
                std::this_thread::sleep_for(std::chrono::milliseconds(NetworkFrequency::subMsgFrequency));
            }
        };
        th_ = std::thread(handler);
        processMsgCb_ = processMsgCb;
        return true;
    }

    Bytes SubscriberNode::Recv()
    {
        auto [zmqMessage, res] = conn_->ReadMsg();
        std::string topic;
        std::string message;
        zmqMessage >> topic >> message;
        return Bytes::from_string(message);
    }
}
